package com.pisolution.scheduler.runtime.worker.processor;

import com.pisolution.scheduler.core.tuple.Tuple3;
import com.pisolution.scheduler.core.utils.Stopper;
import com.pisolution.scheduler.remote.NettyRemoteClient;
import com.pisolution.scheduler.remote.command.Command;
import com.pisolution.scheduler.remote.utils.Host;
import com.pisolution.scheduler.runtime.registry.RegistryCenter;

import io.netty.channel.Channel;
import org.apache.commons.collections.CollectionUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Service;

import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;

/**
 * 用于处理任务回调 主要适用于发送任务的确认与响应指令给管理节点.
 *
 * @author sora
 * @version 0.1.0
 */
@Service
public class TaskCallbackService {
    private static final Logger LOGGER = LoggerFactory.getLogger(TaskCallbackService.class);

    /** 用于确认需要等待的时间. */
    private static final int[] RETRY = {1, 2, 3, 5, 10, 20, 40, 100, 100, 100, 100, 200, 200, 200};

    // -------------------------------------------------------------------------------------------------

    /** 以任务实例标识为主键 存储与该任务实例相关的信道信息. */
    private final Map<Integer, Tuple3<Long, Host, Channel>> channels;

    /** 用于发送交互指令. */
    private final NettyRemoteClient client;

    /** 用于查询管理节点信息. */
    private final RegistryCenter center;

    // -------------------------------------------------------------------------------------------------
    //  初始化
    // -------------------------------------------------------------------------------------------------

    /**
     * 初始化 {@link TaskCallbackService}.
     *
     * @param center 用于查询管理节点信息.
     */
    public TaskCallbackService(final RegistryCenter center) {
        this.channels = new ConcurrentHashMap<>();
        this.client = new NettyRemoteClient();

        this.center = center;
    }

    // -------------------------------------------------------------------------------------------------
    //  属性访问
    // -------------------------------------------------------------------------------------------------

    public void addChannel(final int taskId, final Tuple3<Long, Host, Channel> channel) {
        this.channels.put(taskId, channel);
    }

    // -------------------------------------------------------------------------------------------------
    //  处理回调
    // -------------------------------------------------------------------------------------------------

    /**
     * 发送针对给定任务实例的确认指令.
     *
     * @param taskId 指令所涉及的任务实例标识.
     * @param command 确认指令.
     */
    public void sendAck(final int taskId, final Command command) {
        this.getChannel(taskId).field3.writeAndFlush(command);
    }

    /**
     * 发送针对给定任务实例的响应指令.
     *
     * @param taskId 指令所涉及的任务实例标识.
     * @param command 响应指令.
     */
    public void sendResult(final int taskId, final Command command) {
        this.getChannel(taskId).field3.writeAndFlush(command).addListener(future -> {
            if (future.isSuccess()) {
                this.channels.remove(taskId);
            }
        });
    }

    /**
     * 查找用于给定任务实例的通信信道信息 如果不存在则创建一个新的信道.
     *
     * @param taskId 待查询任务实例标识.
     *
     * @return 该任务实例所对应的通信信道.
     */
    @SuppressWarnings("BusyWait")
    private Tuple3<Long, Host, Channel> getChannel(final int taskId) {
        // 1. 查询给定任务实例的信道信息.
        final Tuple3<Long, Host, Channel> channel = this.channels.get(taskId);

        // 2. 如果信道存在且处于活跃状态则使用此信道 否则创建一个新的信道.
        if (channel != null) {
            if (channel.field3.isActive()) {
                return channel;
            }

            final Channel newChannel = this.client.getChannel(channel.field2);
            if (newChannel != null) {
                final Tuple3<Long, Host, Channel> info = new Tuple3<>(channel.field1, Host.of(newChannel), newChannel);

                this.addChannel(taskId, info);
                return info;
            }
        }

        int retry = 0;
        while (Stopper.isRunning()) {
            // 3. 从注册中心查找所有当前可用管理节点.
            final Set<String> masters = this.center.getMasterServers();

            // 4. 如果无可用管理节点 则等待管理节点可用.
            if (CollectionUtils.isEmpty(masters)) {
                LOGGER.info("Could not find any master server for task {} and try {} times.", taskId, retry + 1);

                try {
                    Thread.sleep(1_000 * RETRY[retry++ % RETRY.length]);
                } catch (InterruptedException ignored) {
                    // 无需处理此异常.
                }

                continue;
            }

            // 5. 如果有面向给定管理节点的通信信道 则使用此信道信息.
            for (final String master: masters) {
                final Channel newChannel = this.client.getChannel(Host.of(master));

                if (newChannel != null) {
                    final Tuple3<Long, Host, Channel> info = new Tuple3<>(-1L, Host.of(newChannel), newChannel);

                    this.addChannel(taskId, info);
                    return info;
                }
            }

            // 6. 等待下一次重试.
            try {
                Thread.sleep(1_000 * RETRY[retry++ % RETRY.length]);
            } catch (InterruptedException ignored) {
                // 无需处理此异常.
            }
        }

        // 7. 如果尝试完所有管理节点依然未能找到可用信道 则判定为状态异常.
        throw new IllegalStateException(String.format("All available master servers are not reachable for task %d", taskId));
    }
}
